package com.hbisdt.cet.sample.consumer;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.stereotype.Service;

/**
 * 碳源销售消息监听器
 * @author : 教学案例
 * @version : 1.0
 * @date : 2025-04-29 15:52
 */
@Service
@RocketMQMessageListener(topic = "CarbonVoucher", consumerGroup = "key_group", selectorExpression = "sell")
public class RocketMQConsumerMessageKey implements RocketMQListener<String> {
    @Override
    public void onMessage(String msgJson) {
        System.out.println("收到原始消息内容：" + msgJson);
        try {
            ObjectMapper mapper = new ObjectMapper();
            JsonNode node = mapper.readTree(msgJson);
            String payload = node.get("payload").asText();
            String key = node.get("headers").get(RocketMQHeaders.KEYS).asText();
            System.out.println("Consumer MessageKey: " + key + ", Received payload: " + payload);

            // 示例：根据消息键进行去重或追踪
            // 在实际场景中，可以使用 key 查询数据库或缓存，检查消息是否已处理
            if (key.startsWith("order_")) {
                System.out.println("Processing order-related message with key: " + key);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
